package com.ustc.hewei.nettyrpc2_0.netty.client.handler;

import com.ustc.hewei.nettyrpc2_0.common.factory.SingletonFactory;
import com.ustc.hewei.nettyrpc2_0.common.warp.RpcRequest;
import com.ustc.hewei.nettyrpc2_0.common.warp.RpcResponse;
import com.ustc.hewei.nettyrpc2_0.netty.client.manager.UnprocessedRequest;
import io.netty.channel.*;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;

/**
* @description: 客户端handler
* @author hewei
* @date 2022/9/7 20:56
* @version 1.0
*/

public class ClientHandler extends SimpleChannelInboundHandler<RpcResponse> {
    private static final Logger logger = LoggerFactory.getLogger(ClientHandler.class);
    private final UnprocessedRequest unprocessedRequest;
    private volatile Channel channel;

    public ClientHandler() {
        unprocessedRequest = SingletonFactory.getSingleton(UnprocessedRequest.class);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        logger.info("客户端已连接 {}", ctx.channel().remoteAddress());
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        super.channelRegistered(ctx);
        channel = ctx.channel();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, RpcResponse response) throws Exception {
        try {
            unprocessedRequest.complete(response);
            logger.info("请求 {} 已收到响应", response.getId());
        } finally {
            ReferenceCountUtil.release(response);
        }
    }

    public CompletableFuture<RpcResponse> sendRequest(RpcRequest request) {
        CompletableFuture<RpcResponse> future = new CompletableFuture<>();
        try {
            unprocessedRequest.add(request.getId(), future);
            channel.writeAndFlush(request).sync().addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (channelFuture.isSuccess()) {
                        if (!request.getHeartBeat()) {
                            logger.info("请求 {} 发送成功", request.getId());
                        }
                    } else {
                        logger.error("请求 {} 发送失败", request.getId());
                        unprocessedRequest.complete(RpcResponse.fail(request,
                                String.format("发送请求 {%S} 失败", request.getId())));
                    }
                }
            });
            return future;
        } catch (InterruptedException e) {
            throw new RuntimeException("发送请求异常", e);
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            channel.writeAndFlush(new RpcRequest().setHeartBeat(true)).sync();
            //logger.info("客户端发送心跳到 {}", ctx.channel().remoteAddress());
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    public void close() {
        channel.close();
        logger.info("{} 已关闭", channel.remoteAddress());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
        logger.error("客户端连接的 {} 服务端异常", ctx.channel().remoteAddress());
    }

    public InetSocketAddress getRemoteAddress() {
        return (InetSocketAddress) channel.remoteAddress();
    }

    public Channel getChannel() {
        return channel;
    }
}